package com.swak.reactivex.cluster.impl;

import static com.swak.Constants.SIDE;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.CompletableFuture;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.Constants;
import com.swak.exception.RemotingException;
import com.swak.reactivex.cluster.Channels;
import com.swak.reactivex.cluster.Cluster;
import com.swak.reactivex.cluster.ID;
import com.swak.reactivex.cluster.Member;
import com.swak.reactivex.cluster.Members;
import com.swak.reactivex.future.Futures;
import com.swak.reactivex.transport.ExchangeChannel;
import com.swak.reactivex.transport.ExchangePeer;
import com.swak.reactivex.transport.ExchangeServer;
import com.swak.reactivex.transport.Message;
import com.swak.utils.Lists;
import com.swak.utils.StringUtils;

public class MemberImpl implements Member {

	private static final Logger LOGGER = LoggerFactory.getLogger(Cluster.class);

	private final ID id;
	private final Cluster cluster;
	private final InetSocketAddress address;
	private volatile boolean closed = false;

	/**
	 * 提供 Udp 单播服务
	 */
	private final ExchangePeer peer;

	/**
	 * 提供 Tcp 可靠服务
	 */
	private final ExchangeServer server;
	private final Channels channels = new Channels();

	MemberImpl(Cluster cluster, ID id) {
		this.id = id;
		this.cluster = cluster;
		this.address = id.getUrl().toInetSocketAddress();
		this.peer = null;
		this.server = null;
	}

	MemberImpl(Cluster cluster, ID id, ExchangeServer server, ExchangePeer peer) {
		this.id = id;
		this.cluster = cluster;
		this.address = id.getUrl().toInetSocketAddress();
		this.peer = peer;
		this.server = server;
	}

	public MemberImpl openChannel(ExchangeChannel channel) {
		this.channels.add(channel);
		return this;
	}

	public MemberImpl closeChannel(ExchangeChannel channel) {
		this.channels.remove(channel);
		if (this.channels.isEmpty()) {
			this.close();
		}
		return this;
	}

	@Override
	public ID id() {
		return id;
	}

	@Override
	public boolean isThis() {
		return this.server != null;
	}

	@Override
	public Members members() {
		return cluster.members();
	}

	@Override
	public Channels channels() {
		return this.channels;
	}

	@Override
	public InetSocketAddress address() {
		return this.address;
	}

	@Override
	public void send(Object message) throws RemotingException {

		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("Send Message Use TCP [{}] to {}", message, this.server != null ? "All But Me" : "One");
		}

		// send all
		if (this.server != null) {
			List<Member> members = cluster.members().members();
			for (Member member : members) {
				if (!member.isThis()) {
					member.send(message);
				}
			}
			return;
		}

		// send one
		ExchangeChannel channel = this.channels.select();
		if (LOGGER.isDebugEnabled()) {
			String side = (String) channel.getAttribute(SIDE);
			LOGGER.debug("Send Message Use TCP-CH side:{}, lo:{}, ra:{}", side, channel.localAddress(),
					channel.remoteAddress());
		}
		channel.send(message);
	}

	@Override
	public CompletableFuture<Object> request(Object request) throws RemotingException {

		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("send message Use TCP {}", request);
		}

		// request all
		if (this.server != null) {
			List<CompletableFuture<Object>> futures = Lists.newArrayList();
			for (ExchangeChannel channel : this.channels.channels()) {
				try {
					futures.add(channel.request(request));
				} catch (Exception e) {
				}
			}
			return Futures.allResult(futures);
		}

		// request one
		return this.channels.select().request(request);
	}

	@Override
	public void broadcast(Object message) throws RemotingException {

		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("Send Message Use UDP [{}] to {}", message, this.server != null ? "All But Me" : "One");
		}

		/*
		 * 如果是组播，则直接发送数据
		 */
		if (this.peer != null && !this.peer.isUnicast()) {
			this.peer.send(Message.forMessage(message));
			return;
		}

		/*
		 * 当前的服务器
		 */
		if (this.isThis() && this.peer != null) {
			List<Member> members = this.members().members();
			for (Member member : members) {
				if (!member.isThis()) {
					Message msg = Message.forMessage(message).setAddress(member.address());
					this.peer.send(msg);
				}
			}
			return;
		}

		/*
		 * 非当前服务器
		 */
		this.broadcast(this.address, message);
	}

	@Override
	public void broadcast(InetSocketAddress address, Object message) throws RemotingException {

		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("Send Message Use UDP [{}] to {}", message, "One");
		}

		/*
		 * 如果是组播，则直接发送数据
		 */
		if (this.peer != null && !this.peer.isUnicast()) {
			this.peer.send(Message.forMessage(message));
			return;
		}

		// 如果是当前的服务则执行发送数据服务
		if (this.isThis() && this.peer != null) {
			Message msg = Message.forMessage(message).setAddress(address);
			this.peer.send(msg);
			return;
		}

		// 找到当前的服务器,执行发送数据的服务
		List<Member> members = this.members().members();
		for (Member member : members) {
			if (member.isThis()) {
				member.broadcast(address, message);
				break;
			}
		}
	}

	@Override
	public CompletableFuture<Object> request(InetSocketAddress address, Object request) throws RemotingException {

		if (LOGGER.isDebugEnabled()) {
			LOGGER.debug("send Message Use TCP {}", request);
		}

		if (this.isThis()) {
			Message msg = Message.forRequest(request).setAddress(address);
			return this.peer.request(msg);
		}

		List<Member> members = this.members().members();
		for (Member member : members) {
			if (member.isThis()) {
				return member.request(address, request);
			}
		}

		return null;
	}

	@Override
	public void close() {
		if (this.closed) {
			return;
		}
		this.closed = true;

		try {
			channels.close();
		} catch (Exception e) {
			LOGGER.error("close channels error.", e);
		}

		try {
			if (this.peer != null) {
				this.peer.close();
			}
		} catch (Exception e) {
			LOGGER.error("close peer error.", e);
		}

		try {
			if (this.server != null) {
				this.server.close();
			}
		} catch (Exception e) {
			LOGGER.error("close server error.", e);
		}

		try {
			if (this.server != null) {
				this.cluster.close();
			}
		} catch (Exception e) {
			LOGGER.error("close cluster error.", e);
		}
	}

	@Override
	public String toString() {
		StringBuilder out = new StringBuilder();
		out.append(this.id.getId()).append("\t");
		out.append(this.id.getUrl().getAddress()).append("\t");
		out.append(this.id.getUrl().getParameter(Constants.Use_Udp, Boolean.FALSE) ? "[TCP、UDP]" : "[TCP]")
				.append("\t");
		out.append(this.server != null ? this.server.getChannels().size() : this.channels.size()).append("ch\t");
		out.append(this.server != null ? "this" : StringUtils.EMPTY);
		return out.toString();
	}
}